package com.gengzp.mq.server.service.impl;

import com.alibaba.nacos.shaded.com.google.common.collect.Lists;
import com.fasterxml.jackson.core.type.TypeReference;
import com.gengzp.mq.common.enums.MessageStatusEnum;
import com.gengzp.mq.common.model.base.MqPaginationResult;
import com.gengzp.mq.common.model.dto.GengzpMqConsumerDto;
import com.gengzp.mq.common.model.entity.GengzpMqMessage;
import com.gengzp.mq.common.model.entity.GengzpMqTopic;
import com.gengzp.mq.common.model.req.PageGengzpMqMessageReq;
import com.gengzp.mq.common.model.resp.GengzpMqMessageExt;
import com.gengzp.mq.common.model.resp.PageGengzpMqMessageResp;
import com.gengzp.mq.common.utils.GengzpMqJsonUtils;
import com.gengzp.mq.common.utils.GengzpMqRedisUtils;
import com.gengzp.mq.common.utils.GengzpMqSnowflakeIdUtils;
import com.gengzp.mq.server.facade.IGengzpMqTopicFacadeService;
import com.gengzp.mq.server.mapper.GengzpMqMessageMapper;
import com.gengzp.mq.server.service.IGengzpMqService;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

import static com.gengzp.mq.common.constants.GengzpMqCacheConstants.*;

/**
 * @ClassName GengzpMqServiceImpl
 * @Description 消息队列模块服务层实现类
 * @Author gengzp
 * @Date 2025/6/8 14:33
 */
@Service
public class GengzpMqServiceImpl implements IGengzpMqService {

    private static final Logger logger = LoggerFactory.getLogger(GengzpMqServiceImpl.class);

    @Autowired
    private GengzpMqMessageMapper gengzpMqMessageMapper;

    @Autowired
    private IGengzpMqTopicFacadeService gengzpMqTopicFacadeService;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Long sendMessage(String topic, String messageContent) {
        // 查询是否存在当前topic
        Object redisTopicInfo = GengzpMqRedisUtils.get(TOPIC_INFO_KEY_PREFIX + topic);
        if (Objects.isNull(redisTopicInfo)) {
            // 缓存中不存在, 则从数据库中查询出topic, 并再次存入缓存
            GengzpMqTopic topicByName = gengzpMqTopicFacadeService.searchTopicByName(topic);
            if (Objects.isNull(topicByName)) {
                // 数据库也不存在topic, 则自动创建topic
                gengzpMqTopicFacadeService.addTopic(topic, "发送消息时, topic不存在, 系统自动创建");
            } else {
                // 数据库中存在, 只是缓存过期了, 只需重新存入缓存
                GengzpMqRedisUtils.set(TOPIC_INFO_KEY_PREFIX + topic,
                        GengzpMqJsonUtils.toJson(topicByName), 1, TimeUnit.DAYS);
            }
        } else {
            // 缓存命中, 续期
            GengzpMqRedisUtils.set(TOPIC_INFO_KEY_PREFIX + topic, redisTopicInfo, 1, TimeUnit.DAYS);
        }

        // 先将消息保存到数据库
        GengzpMqMessage message = new GengzpMqMessage();
        message.setId(GengzpMqSnowflakeIdUtils.nextId());
        message.setMessageKey(String.valueOf(message.getId()));
        message.setTopic(topic);
        message.setMessageContent(messageContent);
        message.setStatus(MessageStatusEnum.UNCONSUMED.getCode()); //未消费状态
        message.setCreatedTime(new Date());
        message.setRetryCount(0);
        gengzpMqMessageMapper.insertMessage(message);

        // 查询所有该topic的消费者组
        Object topicRelationConsumerObj = GengzpMqRedisUtils.get(TOPIC_CONSUMER_SERVICE_NAME_KEY_PREFIX + topic);
        if (Objects.nonNull(topicRelationConsumerObj)) {
            List<GengzpMqConsumerDto> gengzpMqConsumerDtos = GengzpMqJsonUtils.parseObjectWithGeneric(topicRelationConsumerObj.toString(), new TypeReference<>() {
            });
            gengzpMqConsumerDtos.forEach(v -> {
                // 将消息添加到各消费者组队列最右侧
                GengzpMqRedisUtils.rightPush(String.format("%s%s_%s", MESSAGE_KEY_PREFIX, topic, v.getConsumerGroup()),
                        message.getId());
            });
        }else{
            // 将消息添加到默认消费者组队列最右侧
            GengzpMqRedisUtils.rightPush(String.format("%s%s_%s", MESSAGE_KEY_PREFIX, topic, "default"),
                    message.getId());
        }

        return message.getId();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public GengzpMqMessageExt receiveMessage(String consumerServiceIp, String consumerServiceName,
                                             String topic, String consumerGroup) {
        // 保存消费者订阅信息
        gengzpMqTopicFacadeService.saveTopicRelationConsumer(topic, consumerGroup, consumerServiceName,
                consumerServiceIp);

        // 使用 BR POP 命令阻塞获取消息ID, 最大阻塞时间为30秒
        Object messageIdObj = GengzpMqRedisUtils.leftPop(
                String.format("%s%s_%s", MESSAGE_KEY_PREFIX, topic, consumerGroup), 30, TimeUnit.SECONDS);
        if (Objects.isNull(messageIdObj)) {
            return null;
        }

        // 从数据库中查询消息
        GengzpMqMessage message = gengzpMqMessageMapper.selectMessageByMessageKey(
                (Long) messageIdObj);
        if (Objects.isNull(message)) {
            return null;
        }

        // 更新消息状态为消费中
        gengzpMqMessageMapper.updateMessageStatus(message.getId(),
                MessageStatusEnum.CONSUMING.getCode());
        // 将消息id添加到消费中队列
        GengzpMqRedisUtils.rightPush(PROCESSING_MESSAGE_KEY_PREFIX, message.getId());

        GengzpMqMessageExt resp = new GengzpMqMessageExt();
        resp.setId(String.valueOf(message.getId()));
        resp.setMessageKey(message.getMessageKey());
        resp.setTopic(message.getTopic());
        resp.setMessageContent(message.getMessageContent());
        return resp;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void confirmMessage(Long messageId) {
        gengzpMqMessageMapper.updateMessageStatus(messageId, MessageStatusEnum.CONSUMED.getCode());
    }

    @Override
    public MqPaginationResult<PageGengzpMqMessageResp> pageQueryMessage(
            PageGengzpMqMessageReq req) {
        List<GengzpMqMessage> messages = gengzpMqMessageMapper.pageQueryMessage(
                (req.getPageNum() - 1) * req.getPageSize(), req.getPageSize(), req.getTopic(),
                req.getMessageKey(), req.getMessageContent(), req.getStatus());
        if (CollectionUtils.isEmpty(messages)) {
            return MqPaginationResult.get(req.getPageNum(), req.getPageSize(), 0,
                    Collections.emptyList());
        }

        List<PageGengzpMqMessageResp> result = Lists.newArrayList();
        for (GengzpMqMessage message : messages) {
            PageGengzpMqMessageResp resp = new PageGengzpMqMessageResp();
            resp.setId(String.valueOf(message.getId()));
            resp.setMessageKey(message.getMessageKey());
            resp.setTopic(message.getTopic());
            resp.setMessageContent(message.getMessageContent());
            resp.setStatus(message.getStatus());
            resp.setStatusValue(MessageStatusEnum.getValueByCode(message.getStatus()));
            resp.setCreatedTime(message.getCreatedTime());
            resp.setRetryCount(message.getRetryCount());
            result.add(resp);
        }

        return MqPaginationResult.get(req.getPageNum(), req.getPageSize(),
                gengzpMqMessageMapper.selectCountMessage(req.getTopic(), req.getMessageKey(),
                        req.getMessageContent(), req.getStatus()), result);
    }

    @Override
    public MqPaginationResult<PageGengzpMqMessageResp> pageQueryDeathMessage(
            PageGengzpMqMessageReq req) {
        List<GengzpMqMessage> messages = gengzpMqMessageMapper.pageQueryDeathMessage(
                (req.getPageNum() - 1) * req.getPageSize(), req.getPageSize(), req.getTopic(),
                req.getMessageKey(), req.getMessageContent(), req.getStatus());
        if (CollectionUtils.isEmpty(messages)) {
            return MqPaginationResult.get(req.getPageNum(), req.getPageSize(), 0,
                    Collections.emptyList());
        }

        List<PageGengzpMqMessageResp> result = Lists.newArrayList();
        for (GengzpMqMessage message : messages) {
            PageGengzpMqMessageResp resp = new PageGengzpMqMessageResp();
            resp.setId(String.valueOf(message.getId()));
            resp.setMessageKey(message.getMessageKey());
            resp.setTopic(message.getTopic());
            resp.setMessageContent(message.getMessageContent());
            resp.setStatus(message.getStatus());
            resp.setStatusValue(MessageStatusEnum.getValueByCode(message.getStatus()));
            resp.setCreatedTime(message.getCreatedTime());
            resp.setRetryCount(message.getRetryCount());
            result.add(resp);
        }

        return MqPaginationResult.get(req.getPageNum(), req.getPageSize(),
                gengzpMqMessageMapper.selectCountDeathMessage(req.getTopic(), req.getMessageKey(),
                        req.getMessageContent(), req.getStatus()), result);
    }

}
